package com.centsuse.my_redis.config.serializers;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import org.redisson.client.codec.BaseCodec;
import org.redisson.client.handler.State;
import org.redisson.client.protocol.Decoder;
import org.redisson.client.protocol.Encoder;

import java.io.IOException;

/**
 * @author bobo
 */
public class KryoCodec extends BaseCodec {

    private static final ThreadLocal<Kryo> KRYO_THREAD_LOCAL = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.setRegistrationRequired(false);
        kryo.setReferences(true);
        return kryo;
    });

    private final Encoder encoder = new Encoder() {
        @Override
        public ByteBuf encode(Object in) throws IOException {
            ByteBuf out = io.netty.buffer.Unpooled.buffer();
            try (ByteBufOutputStream baos = new ByteBufOutputStream(out);
                 Output output = new Output(baos)) {
                Kryo kryo = KRYO_THREAD_LOCAL.get();
                kryo.writeClassAndObject(output, in);
                output.flush();
                return baos.buffer();
            }
        }
    };

    private final Decoder<Object> decoder = new Decoder<Object>() {
        @Override
        public Object decode(ByteBuf buf, State state) throws IOException {
            try (ByteBufInputStream bais = new ByteBufInputStream(buf);
                 Input input = new Input(bais)) {
                Kryo kryo = KRYO_THREAD_LOCAL.get();
                return kryo.readClassAndObject(input);
            }
        }
    };

    @Override
    public Decoder<Object> getValueDecoder() {
        return decoder;
    }

    @Override
    public Encoder getValueEncoder() {
        return encoder;
    }
}